package jdbcState;

import org.apache.storm.task.IMetricsContext;
import org.apache.storm.trident.state.*;
import org.apache.storm.trident.state.map.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class JDBCState<T> implements IBackingMap<T> {

    private static JDBCStateConfig config;

    JDBCState(JDBCStateConfig config) {
        this.config = config;
    }

    /**
     * 批量的将收到的tuple的key对应的状态从MySQL中取出
     * @param keys
     * @return
     */

    @Override
    public List<T> multiGet(List<List<Object>> keys) {
        // select tel, sum ,txid from  tbl_tel where tel='
        // select name,lng,lat,txid from jing_wei where name='
        StringBuilder sqlBuilder = new StringBuilder("SELECT ").append(config.getName())
                .append("," + config.getLng())
                .append(","+config.getLat())
                .append(",txid")
                .append(" FROM " + config.getTable())
                .append(" WHERE ")
                .append(config.getName())
                .append("='");

        // 构建一个能够连接MySQL的帮助类
        JDBCUtil jdbcUtil = new JDBCUtil(config.getDriver(), config.getUrl(), config.getUsername(), config.getPassword());

        // 用来保存并返回收到的key对应的状态的
        List<Object> result = new ArrayList<Object>();
        Map<String, Object> map = null;

        //  遍历收到的所有的key，将其对应的状态取出来
        for (List<Object> list : keys) {
            // 由于List是一个Tuple的列表，所以，执行get方法，从tuple中的第0个字段获取值
            String key = list.get(3)+"_"+list.get(4);

            // select tel, sum ,txid from  tbl_tel where tel='每一个key的值'
            // 例如：select tel, sum ,txid from  tbl_tel where tel='189111'

            // select name,lng,lat,txid from jing_wei where name='山东_青岛'
            map = jdbcUtil.queryForMap(sqlBuilder.toString() + key + "'");
            System.out.println(sqlBuilder.toString() + key + "'" );

            // 从缓存中获取当前key对应的合计值和上一次更新合计值时的txid
//            Bean itemBean = (Bean) map.get(key);
            areaBean  itemBean=(areaBean) map.get(key);
            long txid = 0L;
//            long val = 0L;
            double lng=0;
            double lat=0;
            if (itemBean != null) {
                lng = itemBean.getLng();
                lat=itemBean.getLat();
                txid = itemBean.getTxid();
            }

            // Opaque类已经帮我们实现了相应的容错性操作，我们这需要将需要更新的sum和txid放入
            // 他们会按照我们学过的txid更新规则来更新我们的sum
            if (config.getType() == StateType.OPAQUE) {
                result.add(new OpaqueValue(txid, lng));
                result.add(new OpaqueValue(txid, lat));
            } else if (config.getType() == StateType.TRANSACTIONAL) {
                result.add(new TransactionalValue(txid, lng));
                result.add(new TransactionalValue(txid, lat));
            } else {
                result.add(lng);
                result.add(lat);
            }
        }
        return (List<T>) result;
    }

    /**
     * 完成的是一次常规的MySQL更新操作，如何去根据txid更新sum值，是Storm的类帮我们完成的计算
     * @param keys
     * @param vals
     */
    @Override
    public void multiPut(List<List<Object>> keys, List<T> vals) {
        //构建新增SQL
        StringBuilder sqlBuilder = new StringBuilder("INSERT INTO ").append(config.getTable())
                .append("(" + config.getName())
                .append("," + config.getLng())
                .append("," + config.getLat())
                .append(",txid")
                .append(") VALUES ");
        for (int i = 0; i < keys.size(); i++) {
            List<Object> key = keys.get(i);
            if (config.getType() == StateType.TRANSACTIONAL) {
                // ***************这一行代码，实现了容错性的计算步骤**************
                TransactionalValue val = (TransactionalValue) vals.get(i);
                sqlBuilder.append("(");
                sqlBuilder.append(key.get(3)+"_"+key.get(4));
                sqlBuilder.append(",");
                sqlBuilder.append(val.getVal());
                sqlBuilder.append(",");
                sqlBuilder.append(val.getTxid());
                sqlBuilder.append(",NOW()");
                sqlBuilder.append("),");
            }
        }
        sqlBuilder.setLength(sqlBuilder.length() - 1);
        System.out.println(sqlBuilder.toString());
        //新增数据
        JDBCUtil jdbcUtil = new JDBCUtil(config.getDriver(), config.getUrl(), config.getUsername(), config.getPassword());
        jdbcUtil.insert(sqlBuilder.toString());
    }

    public static Factory getFactory(JDBCStateConfig config) {
        return new Factory(config);
    }

    static class Factory implements StateFactory {
        private static JDBCStateConfig config;

        public Factory(JDBCStateConfig config) {
            this.config = config;
        }

        @Override
        public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
            final CachedMap map = new CachedMap(new JDBCState(config), config.getCacheSize());
            System.out.println(config);
            if (config.getType() == StateType.OPAQUE) {
                return OpaqueMap.build(map);
            } else if (config.getType() == StateType.TRANSACTIONAL) {
                return TransactionalMap.build(map);
            } else {
                return NonTransactionalMap.build(map);
            }
        }
    }

}
